package org.niit.sql

import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.sql.{DataFrame, SaveMode, SparkSession}

/*
  Spark 连接 MySQL
 */
object SparkSQL_2 {

  def main(args: Array[String]): Unit = {

    val sparkConf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("sparkSQL")
    val spark: SparkSession = SparkSession.builder().config(sparkConf).getOrCreate()
    val sparkContext: SparkContext = spark.sparkContext
    sparkContext.setLogLevel("ERROR")


    //读取数据库
    val dataFrame: DataFrame = spark.read
      .format( "jdbc") //连接MySql JDBC
      .option("url", "jdbc:mysql://node1:3306/BD2") //连接地址 MySQL数据库的版本 8.0+  localhost == 127.0.0.1== 本机IP
      .option("driver", "com.mysql.jdbc.Driver")//数据库驱动  8.0的驱动  com.mysql.cj.jdbc.Driver 5.0：com.mysql.jdbc.Driver
      .option("user", "root")//数据库的用户名
      .option("password", "Niit@123")//数据库的密码
      .option("dbtable", "user")//数据库表
      .load()//加载，将该表里面的数据封装到DataFrame中
    dataFrame.show()


    //写入数据库  dataFrame里面拥有了一些数据
    dataFrame.write
      .format( "jdbc") //连接MySql JDBC
      .option("url", "jdbc:mysql://node1:3306/BD2") //连接地址 MySQL数据库的版本 8.0+  localhost == 127.0.0.1== 本机IP
      .option("driver", "com.mysql.jdbc.Driver")//数据库驱动  8.0的驱动  com.mysql.cj.jdbc.Driver 5.0：com.mysql.jdbc.Driver
      .option("user", "root")//数据库的用户名
      .option("password", "Niit@123")//数据库的密码
      .option("dbtable", "user2")//数据库表
      .mode(SaveMode.Append)//当该表不存在的时候 会自动创建新表
      .save()//加载，将该表里面的数据封装到DataFrame中

    spark.close()
  }

}
